package com.shu.es;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;

public class EsChangeIndexData {
	
	private static SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMddHHmmss");
	private static SimpleDateFormat formatter2 = new SimpleDateFormat("yyyyMM");

	public static void main(String[] args) throws Exception {
		RestHighLevelClient client = getClient("127.0.0.1",9200); //主集群
		String[] indics = {
				"weighthistory_day1",
				"weighthistory_day2018"};
		for(String index : indics) {
			EsChangeIndexData.changeData(client,index);
		}
		
		client.close();
	}
	
	
	public static void changeData(RestHighLevelClient client, String index) throws Exception {
		final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(30L));
		SearchRequest searchRequest = new SearchRequest(index);
		searchRequest.scroll(scroll);
		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
		searchSourceBuilder.size(10000);
		searchRequest.source(searchSourceBuilder);
		SearchResponse searchResponse = client.search(searchRequest); 
		String scrollId = searchResponse.getScrollId();
		SearchHit[] searchHits = searchResponse.getHits().getHits();
		long total = 0;
		String newIndex = "weighthistory_day";

		BulkRequest request = new BulkRequest();
		for (int i = 0; i < searchHits.length; i++) {
			Map<String, Object> map = searchHits[i].getSourceAsMap();
			String tempDateSt = (String)map.get("receiver_time");
			Date oldDate = formatter.parse(tempDateSt);
			String dateSt = formatter2.format(oldDate);
			request.add(new IndexRequest(newIndex+dateSt, searchHits[i].getType(), searchHits[i].getId()).source(map));
			total++;
		}
		request.timeout(TimeValue.timeValueMinutes(20));
		BulkResponse bulkResponse = client.bulk(request);
		if (bulkResponse.hasFailures()) {
			System.out.println(bulkResponse.buildFailureMessage());
		} else {
			System.out.println("========================================" + index + "==" + total + "== > success");
		}
		
		while (searchHits != null && searchHits.length > 0) {
		    SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId); 
		    scrollRequest.scroll(scroll);
		    searchResponse = client.searchScroll(scrollRequest);
		    scrollId = searchResponse.getScrollId();
		    searchHits = searchResponse.getHits().getHits();
		    
		    int size = searchHits.length;
		    request = new BulkRequest();
		    boolean flag = false;
		    for(int i=0; i<size;i++) {
				Map<String, Object> map = searchHits[i].getSourceAsMap();
				String tempDateSt = (String)map.get("receiver_time");
				Date oldDate = formatter.parse(tempDateSt);
				String dateSt = formatter2.format(oldDate);
				request.add(new IndexRequest(newIndex+dateSt, searchHits[i].getType(), searchHits[i].getId()).source(map));
				
				flag = true;
				total ++;
		    }
		    
		    if(flag) {
			    request.timeout(TimeValue.timeValueMinutes(20));
				bulkResponse = client.bulk(request);
				if(bulkResponse.hasFailures()) {
					System.out.println(bulkResponse.buildFailureMessage());
				}else {
					System.out.println("========================================"+index+"=="+total+"== > success");
				}
		    }
		}
		System.out.println("---------------"+total);
	}
	
	
	public static RestHighLevelClient getClient(String ip,Integer port) {
		List<HttpHost> list = new ArrayList<HttpHost>();
		list.add(new HttpHost(ip, port, "http"));
		HttpHost[] arr = new HttpHost[list.size()];
		arr = list.toArray(arr);
		
		RestClientBuilder builder = RestClient.builder(arr)
				.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
					public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
						requestConfigBuilder.setConnectTimeout(5000);
						requestConfigBuilder.setSocketTimeout(40000);
						requestConfigBuilder.setConnectionRequestTimeout(1000);
						return requestConfigBuilder;
					}

				}).setMaxRetryTimeoutMillis(20 * 60 * 1000);

		return new RestHighLevelClient(builder);
	}
}
